package com.lagou.cepstudy;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

public class OutOfOder {
    public static void main(String[] args) throws Exception {
        //运行上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        DataStreamSource<ChengkeBean> data = env.fromElements(
                new ChengkeBean(1, "create", 1597905234000L),
                new ChengkeBean(1, "ok", 1597905235000L),
                new ChengkeBean(2, "create", 1597905236000L),
                new ChengkeBean(2, "ok", 1597905237000L),
                new ChengkeBean(3, "create", 1597905239000L),
                new ChengkeBean(4, "create", 1597905238000L)
        );

        //watermark  --水印 ： 允许产生乱序事件
        SingleOutputStreamOperator<ChengkeBean> watermarks = data.assignTimestampsAndWatermarks(new WatermarkStrategy<ChengkeBean>() {
            @Override
            public WatermarkGenerator<ChengkeBean> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<ChengkeBean>() {
                    long maxTimeStamp = Long.MIN_VALUE;

                    @Override
                    public void onEvent(ChengkeBean event, long eventTimestamp, WatermarkOutput output) {
                        maxTimeStamp = Math.max(maxTimeStamp, event.getTs());
                    }

                    long maxOutofOrderness = 500l;

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        output.emitWatermark(new Watermark(maxTimeStamp - maxOutofOrderness));
                    }
                };
            }
        }.withTimestampAssigner((element, recordTimestamp) -> element.getTs()));

        KeyedStream<ChengkeBean, Integer> keyed = watermarks.keyBy(value -> value.getId());

        //定义模式
        Pattern<ChengkeBean, ChengkeBean> pattern = Pattern.<ChengkeBean>begin("start").where(new IterativeCondition<ChengkeBean>() {
            @Override
            public boolean filter(ChengkeBean value, Context<ChengkeBean> ctx) throws Exception {
                return value.getState().equals("create");
            }
        })
                .followedBy("next").where(new IterativeCondition<ChengkeBean>() {
                    @Override
                    public boolean filter(ChengkeBean value, Context<ChengkeBean> ctx) throws Exception {
                        return value.getState().equals("ok");
                    }
                })
                .within(Time.seconds(60 * 2));

        PatternStream<ChengkeBean> patternStream = CEP.pattern(keyed, pattern);

        OutputTag<ChengkeBean> outputTag = new OutputTag<ChengkeBean>("outoftime") {
        };
        SingleOutputStreamOperator<Integer> result = patternStream.select(outputTag, new PatternTimeoutFunction<ChengkeBean, ChengkeBean>() {
            @Override
            public ChengkeBean timeout(Map<String, List<ChengkeBean>> pattern, long timeoutTimestamp) throws Exception {
                return pattern.get("start").get(0);
            }
        }, new PatternSelectFunction<ChengkeBean, Integer>() {
            @Override
            public Integer select(Map<String, List<ChengkeBean>> pattern) throws Exception {
                return pattern.get("next").get(0).getId();
            }
        });

        DataStream<ChengkeBean> sideOutput = result.getSideOutput(outputTag);
        sideOutput.print();

        env.execute();

    }
}
